package jp.co.nes.awssupport.common.utility;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.StringUtils;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBHashKey;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBMapper;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBQueryExpression;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBRangeKey;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBScanExpression;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBTable;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBTypeConverted;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.dynamodbv2.util.TableUtils;
import com.amazonaws.services.dynamodbv2.util.TableUtils.TableNeverTransitionedToStateException;

import jp.co.nes.awssupport.common.base.CommonException;
import jp.co.nes.awssupport.common.bean.BaseBean;

/**
 * DynamoDBヘルパーです。<br>
 * DynamoDBに関する機能を提供します。
 */
public class AwsDynamoDBHelper extends AwsHelperBase {

    /** クライアント */
    private AmazonDynamoDB client;

    /**
     * デフォルトコンストラクタです。
     */
    public AwsDynamoDBHelper() {
    }

    /*
     * (非 Javadoc)
     * @see jp.co.nes.awssupport.common.utility.AwsHelperBase#init(com.amazonaws.regions.Regions, com.amazonaws.auth.AWSCredentials, com.amazonaws.ClientConfiguration)
     */
    @Override
    protected void init(Regions regions, AWSCredentials credentials, ClientConfiguration config) {
        boolean isLocalTest = Boolean.valueOf(PropertiesUtil.getProperty("LocalTest", "false"));
        if (isLocalTest) {
            // DynamoDBローカル用のクライアントを生成する。
            String dummyAccessKey = "dummy";
            String dummySecretKey = "dummy";
            String dynamoDBLocalEndpoint = PropertiesUtil.getProperty("DynamoDBLocalEndpoint", "http://localhost:8000");
            String dynamoDBLocalRegion = PropertiesUtil.getProperty("DynamoDBLocalRegion", "ap-northeast-1");
            client = AmazonDynamoDBClientBuilder.standard()
                    .withCredentials(
                            new AWSStaticCredentialsProvider(createAWSCredentials(dummyAccessKey, dummySecretKey)))
                    .withEndpointConfiguration(
                            new AwsClientBuilder.EndpointConfiguration(dynamoDBLocalEndpoint, dynamoDBLocalRegion))
                    .build();
        } else {
            if (credentials == null) {
                if (config == null) {
                    client = AmazonDynamoDBClientBuilder.standard()
                            .withRegion(regions)
                            .build();
                } else {
                    client = AmazonDynamoDBClientBuilder.standard()
                            .withRegion(regions)
                            .withClientConfiguration(config)
                            .build();
                }
            } else {
                if (config == null) {
                    client = AmazonDynamoDBClientBuilder.standard()
                            .withRegion(regions)
                            .withCredentials(new AWSStaticCredentialsProvider(credentials))
                            .withClientConfiguration(new ClientConfiguration())
                            .build();
                } else {
                    client = AmazonDynamoDBClientBuilder.standard()
                            .withRegion(regions)
                            .withCredentials(new AWSStaticCredentialsProvider(credentials))
                            .withClientConfiguration(config)
                            .build();
                }
            }
        }
    }

    /**
     * テーブルを作成します。
     *
     * @param cls テーブルを作成するBeanクラスのClassオブジェクト
     * @param rc ReadCapacityUnits
     * @param wc WriteCapacityUnits
     */
    public <T extends BaseBean> void createTable(Class<T> cls, long rc, long wc) {
        // DynamoDB関連のアノテーションからテーブル名とパーティションキー、ソートキーの内容を取得する。
        DynamoDBTable dynamoDBTable = cls.getDeclaredAnnotation(DynamoDBTable.class);
        String tableName = dynamoDBTable.tableName();
        String paritionKey = null;
        Class<?> paritionKeyType = null;
        String sortKey = null;
        Class<?> sortKeyType = null;
        Method[] methods = cls.getMethods();
        for (Method method : methods) {
            DynamoDBHashKey dynamoDBHashKey = method.getDeclaredAnnotation(DynamoDBHashKey.class);
            DynamoDBRangeKey dynamoDBRangeKey = method.getDeclaredAnnotation(DynamoDBRangeKey.class);
            DynamoDBTypeConverted dynamoDBTypeConverted = method.getDeclaredAnnotation(DynamoDBTypeConverted.class);
            if (dynamoDBHashKey != null) {
                // パーティションキーの名前と型のClassオブジェクトを取得する。
                String attributeName = dynamoDBHashKey.attributeName();
                if (StringUtils.isEmpty(attributeName)) {
                    // アノテーションのattributeNameに値が設定されていない場合は、メソッド名を使用する。
                    paritionKey = getKey(method.getName());
                } else {
                    paritionKey = attributeName;
                }
                paritionKeyType = getKeyType(method.getReturnType(), dynamoDBTypeConverted);
            }
            if (dynamoDBRangeKey != null) {
                // ソートキーの名前と型のClassオブジェクトを取得する。
                String attributeName = dynamoDBRangeKey.attributeName();
                if (StringUtils.isEmpty(attributeName)) {
                    // アノテーションのattributeNameに値が設定されていない場合は、メソッド名を使用する。
                    sortKey = getKey(method.getName());
                } else {
                    sortKey = attributeName;
                }
                sortKeyType = getKeyType(method.getReturnType(), dynamoDBTypeConverted);
            }
        }
        createTable(tableName, paritionKey, paritionKeyType, sortKey, sortKeyType, rc, wc);
    }

    /**
     * テーブルを作成します。
     *
     * @param tableName テーブル名
     * @param paritionKey パーティションキーの名前
     * @param paritionKeyType パーティションキーの型のClassオブジェクト
     * @param sortKey ソートキーの名前
     * @param sortKeyType ソートキーの型のClassオブジェクト
     * @param rc ReadCapacityUnits
     * @param wc WriteCapacityUnits
     */
    public void createTable(String tableName, String paritionKey, Class<?> paritionKeyType, String sortKey,
            Class<?> sortKeyType, long rc, long wc) {
        CreateTableRequest request = new CreateTableRequest();
        request.setTableName(tableName);

        List<KeySchemaElement> kseList = new ArrayList<>();
        kseList.add(new KeySchemaElement().withAttributeName(paritionKey).withKeyType(KeyType.HASH));
        if (sortKey != null) {
            kseList.add(new KeySchemaElement().withAttributeName(sortKey).withKeyType(KeyType.RANGE));
        }
        request.setKeySchema(kseList);

        List<AttributeDefinition> adList = new ArrayList<>();
        ScalarAttributeType paritionKeySaType = getScalarAttributeType(paritionKeyType);
        adList.add(new AttributeDefinition().withAttributeName(paritionKey).withAttributeType(paritionKeySaType));
        if (sortKey != null) {
            ScalarAttributeType sortKeySaType = getScalarAttributeType(sortKeyType);
            adList.add(new AttributeDefinition().withAttributeName(sortKey).withAttributeType(sortKeySaType));
        }
        request.setAttributeDefinitions(adList);

        ProvisionedThroughput pt = new ProvisionedThroughput().withReadCapacityUnits(rc).withWriteCapacityUnits(wc);
        request.setProvisionedThroughput(pt);

        // テーブルがまだ作成されていない場合に作成する。
        TableUtils.createTableIfNotExists(client, request);

        try {
            // テーブルがアクティブになるまで待つ。
            TableUtils.waitUntilActive(client, tableName);
        } catch (TableNeverTransitionedToStateException | InterruptedException e) {
            throw new CommonException(e);
        }
    }

    /**
     * スキャンを実行します。
     *
     * @param cls スキャンを実行するテーブルに対応するBeanクラスのClassオブジェクト
     * @param dynamoDBScanExpression スキャンを実行する際に引き渡す内容
     * @return 実行結果のリスト
     */
    public <T extends BaseBean> List<T> scan(Class<T> cls, DynamoDBScanExpression dynamoDBScanExpression) {
        DynamoDBMapper mapper = new DynamoDBMapper(client);
        return mapper.scan(cls, dynamoDBScanExpression);
    }

    /**
     * クエリーを実行します。
     *
     * @param cls クエリーを実行するテーブルに対応するBeanクラスのClassオブジェクト
     * @param dynamoDBQueryExpression クエリーを実行する際に引き渡す内容
     * @return 実行結果のリスト
     */
    public <T extends BaseBean> List<T> query(Class<T> cls, DynamoDBQueryExpression<T> dynamoDBQueryExpression) {
        DynamoDBMapper mapper = new DynamoDBMapper(client);
        return mapper.query(cls, dynamoDBQueryExpression);
    }

    /**
     * Beanクラスの内容を保存します。
     *
     * @param bean Beanクラス
     */
    public <T extends BaseBean> void save(T bean) {
        DynamoDBMapper mapper = new DynamoDBMapper(client);
        mapper.save(bean);
    }

    /**
     * Beanクラスの内容を削除します。
     *
     * @param bean Beanクラス
     */
    public <T extends BaseBean> void delete(T bean) {
        DynamoDBMapper mapper = new DynamoDBMapper(client);
        mapper.delete(bean);
    }

    /**
     * Getterメソッド名から先頭のgetを除いた文字列の先頭を小文字にした文字列を返却します。
     *
     * @param getterMethodName Getterメソッド名
     * @return Getterメソッド名から先頭のgetを除いた文字列の先頭を小文字にした文字列
     */
    private String getKey(String getterMethodName) {
        String value = getterMethodName.substring("get".length());
        String firstValue = value.substring(0, 1);
        return firstValue.toLowerCase() + value.substring(1);
    }

    /**
     * 型を返却します。<br>
     * dynamoDBTypeConvertedがnullの場合は、typeをそのまま返却します。<br>
     * dynamoDBTypeConvertedがnullでない場合は、dynamoDBTypeConvertedを使用して、コンバートした型を返却します。
     *
     * @param type 型をあらわすクラスオブジェクト
     * @param dynamoDBTypeConverted コンバータ
     * @return 型
     */
    private Class<?> getKeyType(Class<?> type, DynamoDBTypeConverted dynamoDBTypeConverted) {
        Class<?> keyType = null;
        if (dynamoDBTypeConverted == null) {
            keyType = type;
        } else {
            Class<?> converter = dynamoDBTypeConverted.converter();
            Method[] methods = converter.getMethods();
            for (Method method : methods) {
                if ("convert".equals(method.getName())) {
                    keyType = method.getReturnType();
                }
            }
        }
        if (keyType == null) {
            throw new CommonException("keyType is null");
        }
        return keyType;
    }

    /**
     * typeに対応するScalarAttributeTypeを返却します。
     *
     * @param type 型をあらわすクラスオブジェクト
     * @return typeに対応するScalarAttributeTypeを
     */
    private ScalarAttributeType getScalarAttributeType(Class<?> type) {
        ScalarAttributeType result = null;
        if (type.isAssignableFrom(String.class)) {
            result = ScalarAttributeType.S;
        } else if (type.isAssignableFrom(int.class)) {
            result = ScalarAttributeType.N;
        } else {
            // 現状はString, intのみに対応(他のものが必要になった際は、そのときに実装)
            throw new CommonException("Type is wrong");
        }
        return result;
    }
}
